package com.ecommerce.user.service.impl;

import com.ecommerce.common.exception.BusinessException;
import com.ecommerce.common.util.JwtUtil;
import com.ecommerce.user.config.KafkaConfig;
import com.ecommerce.common.event.UserRegisteredEvent;
import com.ecommerce.common.event.UserDeletedEvent;
import com.ecommerce.user.controller.UserController;
import com.ecommerce.user.dto.*;
import com.ecommerce.user.entity.User;
import com.ecommerce.user.dto.UserMapper;
import com.ecommerce.user.repository.UserRepository;
import com.ecommerce.user.service.UserService;
import com.ecommerce.user.service.UserEventPublisher;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.Pageable;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.security.core.Authentication;
import org.springframework.security.core.context.SecurityContextHolder;
import org.springframework.security.core.authority.SimpleGrantedAuthority;
import org.springframework.security.crypto.bcrypt.BCryptPasswordEncoder;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.context.request.RequestContextHolder;
import org.springframework.web.context.request.ServletRequestAttributes;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.core.ParameterizedTypeReference;
import reactor.core.publisher.Mono;

import jakarta.servlet.http.HttpServletRequest;

import java.time.LocalDateTime;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
// 移除未使用的 import java.util.Optional
import java.util.Set;
import java.util.stream.Collectors;
import java.time.Duration;

/**
 * 用户服务实现类
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class UserServiceImpl implements UserService {

    private final UserRepository userRepository;
    private final UserMapper userMapper;
    private final BCryptPasswordEncoder passwordEncoder;
    private final WebClient webClient;
    private final RestTemplate restTemplate;
    private final KafkaConfig.UserRegisteredEventProducer userRegisteredEventProducer;
    private final KafkaConfig.UserDeletedEventProducer userDeletedEventProducer;
    private final UserEventPublisher userEventPublisher;

    @Value("${auth.service.url:http://localhost:8084}")
    private String authServiceUrl;

    @Override
    @Transactional
    public UserResponse register(UserRegisterRequest registerRequest) {
        log.info("用户注册请求: {}", registerRequest.getUsername());

        // 首先检查用户是否已存在
        if (userRepository.existsByUsername(registerRequest.getUsername())) {
            log.warn("用户服务中用户名已存在: {}", registerRequest.getUsername());
            return getUserByUsername(registerRequest.getUsername());
        }

        if (userRepository.existsByEmail(registerRequest.getEmail())) {
            log.warn("用户服务中邮箱已存在: {}", registerRequest.getEmail());
            throw new BusinessException("邮箱已被注册");
        }

        // 调用认证服务注册接口
        try {
            Map<String, Object> authRegisterRequest = new HashMap<>();
            authRegisterRequest.put("username", registerRequest.getUsername());
            authRegisterRequest.put("password", registerRequest.getPassword());
            authRegisterRequest.put("email", registerRequest.getEmail());

            // 调用认证服务注册接口
            Map<String, Object> authResponse;
            try {
                authResponse = webClient.post()
                        .uri("/api/auth/register")
                        .contentType(MediaType.APPLICATION_JSON)
                        .bodyValue(authRegisterRequest)
                        .retrieve()
                        .bodyToMono(new ParameterizedTypeReference<Map<String, Object>>() {})
                        .block();
            } catch (Exception e) {
                log.error("调用认证服务注册接口失败: {}", e.getMessage());
                throw new BusinessException("注册服务暂时不可用，请稍后重试");
            }

            log.info("认证服务注册成功: {}", authResponse);
            
            // 从认证服务响应中获取用户ID
            Long authUserId = null;
            if (authResponse != null) {
                // 尝试从data字段中获取AuthResponse对象
                if (authResponse.containsKey("data")) {
                    Map<String, Object> data = (Map<String, Object>) authResponse.get("data");
                    if (data != null && data.containsKey("userId")) {
                        authUserId = Long.valueOf(data.get("userId").toString());
                        log.info("从认证服务响应data中获取到用户ID: {}", authUserId);
                    }
                }
                // 如果data中没有，尝试直接从响应中获取
                else if (authResponse.containsKey("userId")) {
                    authUserId = Long.valueOf(authResponse.get("userId").toString());
                    log.info("直接从认证服务响应中获取到用户ID: {}", authUserId);
                }
            }
            
            // 认证服务注册成功后，在用户服务中创建用户记录
            User user = new User();
            // 如果认证服务返回了用户ID，使用认证服务的ID，否则使用数据库自动生成的ID
            if (authUserId != null) {
                user.setId(authUserId);
                log.info("设置用户ID为认证服务返回的ID: {}", authUserId);
            } else {
                log.warn("认证服务未返回用户ID，将使用数据库自动生成的ID");
            }
            user.setUsername(registerRequest.getUsername());
            user.setPassword(""); // 密码由认证服务管理，用户服务不需要存储
            user.setEmail(registerRequest.getEmail());
            user.setRoles(Collections.singleton("USER"));

            // 保存用户并确保创建成功
            user = userRepository.save(user);
            
            // 验证用户是否成功创建
            if (user.getId() == null) {
                log.error("用户服务数据创建失败，用户ID为空: {}", registerRequest.getUsername());
                throw new BusinessException("用户数据创建失败，请重试");
            }
            
            // 双重确认用户数据已持久化
            final String username = registerRequest.getUsername();
            final Long userId = user.getId();
            User savedUser = userRepository.findById(userId)
                    .orElseThrow(() -> {
                        log.error("用户数据保存后验证失败: userId={}, username={}", 
                                 userId, username);
                        return new BusinessException("用户数据保存验证失败");
                    });
            
            log.info("用户服务注册成功: {}, userId: {}", savedUser.getUsername(), savedUser.getId());

            // 发送用户注册成功事件到Kafka
            try {
                UserRegisteredEvent event = UserRegisteredEvent.builder()
                        .userId(savedUser.getId())
                        .username(savedUser.getUsername())
                        .email(savedUser.getEmail())
                        .roles(new HashSet<>(savedUser.getRoles()))
                        .registerTime(LocalDateTime.now())
                        .build();
                
                // 使用新的用户事件发布器（主要方式）
                try {
                    userEventPublisher.publishUserCreated(savedUser);
                    log.info("用户创建事件已通过新发布器发送 - 用户ID: {}, 用户名: {}", savedUser.getId(), savedUser.getUsername());
                } catch (Exception publisherException) {
                    log.warn("新用户事件发布器发送失败，回退到原Kafka发送 - 用户ID: {}, 错误: {}", 
                             savedUser.getId(), publisherException.getMessage());
                    // 回退到原来的Kafka发送方式
                    userRegisteredEventProducer.sendMessage(event);
                    log.info("用户注册成功事件已发送到Kafka(回退方式) - 用户ID: {}, 用户名: {}", savedUser.getId(), savedUser.getUsername());
                }
                
            } catch (Exception e) {
                log.error("用户注册同步过程失败 - 用户ID: {}, 错误: {}", savedUser.getId(), e.getMessage(), e);
                
                // 清理已创建的用户数据
                try {
                    userRepository.delete(savedUser);
                    log.info("注册失败，已清理用户数据 - 用户ID: {}", savedUser.getId());
                } catch (Exception cleanupEx) {
                    log.error("清理用户数据失败 - 用户ID: {}, 错误: {}", savedUser.getId(), cleanupEx.getMessage());
                }
                
                throw new BusinessException("注册失败: " + e.getMessage());
            }

            return userMapper.entityToResponseDto(savedUser);
            
        } catch (Exception e) {
            log.error("注册过程失败: {}", e.getMessage(), e);
            
            // 如果是用户服务数据创建失败，需要回滚认证服务注册
            if (e.getMessage() != null && (e.getMessage().contains("用户数据创建失败") || 
                                         e.getMessage().contains("用户数据保存验证失败"))) {
                log.warn("用户服务数据创建失败，准备回滚认证服务注册: {}", registerRequest.getUsername());
                
                try {
                    // 调用认证服务删除用户接口进行回滚
                    final String rollbackUsername = registerRequest.getUsername();
                    webClient.delete()
                            .uri("/api/auth/user/" + rollbackUsername)
                            .retrieve()
                            .bodyToMono(Void.class)
                            .onErrorResume(rollbackError -> {
                                log.error("回滚认证服务注册失败: username={}, error={}", 
                                         rollbackUsername, rollbackError.getMessage());
                                return Mono.empty();
                            })
                            .block();
                    
                    log.info("认证服务注册回滚成功: {}", rollbackUsername);
                } catch (Exception rollbackEx) {
                    log.error("认证服务注册回滚失败: username={}, error={}", 
                             registerRequest.getUsername(), rollbackEx.getMessage(), rollbackEx);
                }
                
                throw new BusinessException("注册失败，请重试");
            }
            
            // 其他异常处理
            if (e.getMessage() != null && e.getMessage().contains("用户名已存在")) {
                throw new BusinessException("用户名已存在");
            } else if (e.getMessage() != null && e.getMessage().contains("邮箱已被注册")) {
                throw new BusinessException("邮箱已被注册");
            } else {
                throw new BusinessException("注册失败: " + e.getMessage());
            }
        }
    }

    @Override
    public UserLoginResponse login(UserLoginRequest loginRequest) {
        log.info("用户登录请求: {}", loginRequest.getUsername());

        // 首先调用认证服务验证用户名和密码
        try {
            Map<String, Object> authLoginRequest = new HashMap<>();
            authLoginRequest.put("username", loginRequest.getUsername());
            authLoginRequest.put("password", loginRequest.getPassword());

            // 调用认证服务登录接口
            Map<String, Object> authResponse;
            try {
                authResponse = webClient.post()
                        .uri("/api/auth/login")
                        .contentType(MediaType.APPLICATION_JSON)
                        .bodyValue(authLoginRequest)
                        .retrieve()
                        .bodyToMono(new ParameterizedTypeReference<Map<String, Object>>() {})
                        .block();
            } catch (Exception e) {
                log.error("调用认证服务登录接口失败: {}", e.getMessage());
                throw new BusinessException("登录服务暂时不可用，请稍后重试");
            }

            log.info("认证服务登录成功: {}", authResponse);
            
            // 认证服务登录成功后，获取用户信息
            User user = userRepository.findByUsername(loginRequest.getUsername())
                    .orElseThrow(() -> new BusinessException("用户不存在"));

            // 从认证服务响应中获取token
            String token = (String) authResponse.get("token");
            if (token == null) {
                // 如果认证服务没有返回token，自己生成一个
                List<String> rolesList = user.getRoles().stream().collect(Collectors.toList());
                token = JwtUtil.generateToken(user.getUsername(), rolesList);
            }
            
            // 构建响应
            UserResponse userResponse = userMapper.entityToResponseDto(user);
            UserLoginResponse loginResponse = new UserLoginResponse(token, userResponse);
            
            log.info("用户登录成功: {}", user.getUsername());
            return loginResponse;
            
        } catch (Exception e) {
            log.error("调用认证服务登录失败: {}", e.getMessage());
            throw new BusinessException("用户名或密码错误");
        }
    }

    @Override
    public UserResponse getUserById(Long userId) {
        log.info("根据ID获取用户信息: {}", userId);

        // 权限验证 - 确保用户只能访问自己的信息
        Authentication authentication = SecurityContextHolder.getContext().getAuthentication();
        if (authentication != null && authentication.isAuthenticated() && !authentication.getPrincipal().equals("anonymousUser")) {
            String currentUsername = authentication.getName();
            
            // 尝试通过用户名获取用户信息
            User currentUser = null;
            try {
                currentUser = userRepository.findByUsername(currentUsername)
                        .orElseThrow(() -> new BusinessException("当前用户不存在"));
            } catch (BusinessException e) {
                // 如果通过用户名找不到用户，可能是网关传递的用户信息
                log.warn("通过用户名{}找不到用户，可能是因为用户信息来自网关，跳过数据库用户验证", currentUsername);
            }
            
            // 检查是否是管理员或访问自己的信息
            boolean isAdmin = authentication.getAuthorities().stream()
                    .anyMatch(auth -> auth.getAuthority().equals("ROLE_ADMIN"));
            
            // 如果找到了当前用户，检查ID是否匹配
            if (currentUser != null) {
                log.info("当前用户: {}, ID: {}, 请求用户ID: {}", currentUsername, currentUser.getId(), userId);
                if (!isAdmin && !currentUser.getId().equals(userId)) {
                    log.warn("权限验证失败: 用户 {} 尝试访问用户 {} 的信息", currentUser.getId(), userId);
                    throw new BusinessException("无权访问其他用户信息");
                }
            } else {
                // 如果找不到当前用户，但用户已认证，可能是网关传递的用户信息
                // 这种情况下，我们信任网关的认证结果，允许用户访问信息
                log.info("网关用户访问个人信息，用户名: {}, 请求用户ID: {}", currentUsername, userId);
                
                // 从认证信息中获取用户ID进行比较
                Object principal = authentication.getPrincipal();
                if (principal instanceof org.springframework.security.core.userdetails.User) {
                    org.springframework.security.core.userdetails.User userDetails = 
                            (org.springframework.security.core.userdetails.User) principal;
                    // 检查用户ID是否匹配
                    if (!isAdmin && !userDetails.getUsername().equals(userId.toString())) {
                        log.warn("网关用户权限验证失败: 用户 {} 尝试访问用户 {} 的信息", userDetails.getUsername(), userId);
                        throw new BusinessException("无权访问其他用户信息");
                    }
                }
            }
        }

        User user = userRepository.findById(userId)
                .orElse(null);
        
        // 如果用户不存在，尝试从认证服务同步用户信息
        if (user == null) {
            log.info("用户 {} 在用户服务中不存在，尝试从认证服务同步", userId);
            try {
                // 使用syncUserFromAuthService方法同步用户数据
                syncUserFromAuthService(userId);
                
                // 重新获取用户
                user = userRepository.findById(userId)
                        .orElseThrow(() -> new BusinessException("同步用户数据后仍无法找到用户"));
                
                log.info("用户数据同步成功: {}", userId);
            } catch (Exception e) {
                log.error("同步用户数据失败: userId={}, error={}", userId, e.getMessage(), e);
                throw new BusinessException("用户不存在且同步失败: " + e.getMessage());
            }
        }
        
        // 获取响应DTO并确保字段名兼容性
        UserResponse response = userMapper.entityToResponseDto(user);
        
        // 设置别名字段以确保兼容性
        response.setUserId(response.getId()); // 设置userId别名
        response.setPhoneNumber(response.getPhone()); // 设置phoneNumber别名
        response.setAvatarUrl(response.getAvatar()); // 设置avatarUrl别名
        response.setNickName(response.getNickname()); // 设置nickName别名
        
        log.info("用户信息获取成功: {}", user.getUsername());
        return response;
    }

    @Override
    public UserResponse getUserByUsername(String username) {
        log.info("根据用户名获取用户信息: {}", username);

        User user = userRepository.findByUsername(username)
                .orElseThrow(() -> new BusinessException("用户不存在"));
        
        return userMapper.entityToResponseDto(user);
    }

    @Override
    @Transactional
    public UserResponse updateProfile(Long userId, UpdateProfileRequest updateRequest) {
        log.info("更新用户信息: {}", userId);

        // 权限验证 - 确保用户只能更新自己的信息
        Authentication authentication = SecurityContextHolder.getContext().getAuthentication();
        if (authentication != null && authentication.isAuthenticated() && !authentication.getPrincipal().equals("anonymousUser")) {
            String currentUsername = authentication.getName();
            
            // 尝试通过用户名获取用户信息
            User currentUser = null;
            try {
                currentUser = userRepository.findByUsername(currentUsername)
                        .orElseThrow(() -> new BusinessException("当前用户不存在"));
            } catch (BusinessException e) {
                // 如果通过用户名找不到用户，可能是网关传递的用户信息
                log.warn("通过用户名{}找不到用户，可能是因为用户信息来自网关，跳过数据库用户验证", currentUsername);
            }
            
            // 检查是否是管理员或更新自己的信息
            boolean isAdmin = authentication.getAuthorities().stream()
                    .anyMatch(auth -> auth.getAuthority().equals("ROLE_ADMIN"));
            
            // 如果找到了当前用户，检查ID是否匹配
            if (currentUser != null) {
                if (!isAdmin && !currentUser.getId().equals(userId)) {
                    throw new BusinessException("无权更新其他用户信息");
                }
            } else {
                // 如果找不到当前用户，但用户已认证，可能是网关传递的用户信息
                // 这种情况下，我们信任网关的认证结果，但需要验证用户ID
                log.info("网关用户更新资料，用户名: {}, 请求用户ID: {}", currentUsername, userId);
                
                // 从认证信息中获取用户ID进行比较
                Object principal = authentication.getPrincipal();
                if (principal instanceof org.springframework.security.core.userdetails.User) {
                    org.springframework.security.core.userdetails.User userDetails = 
                            (org.springframework.security.core.userdetails.User) principal;
                    // 检查用户ID是否匹配
                    if (!isAdmin && !userDetails.getUsername().equals(userId.toString())) {
                        log.warn("网关用户权限验证失败: 用户 {} 尝试修改用户 {} 的资料", userDetails.getUsername(), userId);
                        throw new BusinessException("无权更新其他用户信息");
                    }
                }
            }
        }

        User user = userRepository.findById(userId)
                .orElseThrow(() -> new BusinessException("用户不存在"));

        // 检查邮箱是否被其他用户使用
        if (updateRequest.getEmail() != null && !updateRequest.getEmail().equals(user.getEmail())) {
            if (userRepository.existsByEmailAndIdNot(updateRequest.getEmail(), userId)) {
                throw new BusinessException("邮箱已被其他用户使用");
            }
            user.setEmail(updateRequest.getEmail());
        }

        // 更新其他字段
        if (updateRequest.getPhone() != null) {
            user.setPhone(updateRequest.getPhone());
        }
        if (updateRequest.getAddress() != null) {
            user.setAddress(updateRequest.getAddress());
        }
        if (updateRequest.getNickname() != null) {
            user.setNickname(updateRequest.getNickname());
        }
        if (updateRequest.getAvatar() != null) {
            user.setAvatar(updateRequest.getAvatar());
        }

        // 先更新用户服务数据库
        user = userRepository.save(user);
        log.info("用户服务数据库更新成功: {}", userId);
        
        // 发布用户更新事件到Kafka
        try {
            userEventPublisher.publishUserUpdated(user);
            log.info("用户更新事件已发布到Kafka - 用户ID: {}", userId);
        } catch (Exception eventException) {
            log.warn("用户更新事件发布失败 - 用户ID: {}, 错误: {}", userId, eventException.getMessage());
        }
        
        // 同步更新认证服务数据库
        try {
            String url = authServiceUrl + "/api/auth/update-profile/" + userId;
            
            HttpHeaders headers = new HttpHeaders();
            headers.setContentType(MediaType.APPLICATION_JSON);
            
            // 构建请求体，注意字段名映射
            Map<String, Object> request = new HashMap<>();
            request.put("nickName", updateRequest.getNickname());  // 用户服务的nickname对应认证服务的nickName
            request.put("phoneNumber", updateRequest.getPhone());  // 用户服务的phone对应认证服务的phoneNumber
            request.put("email", updateRequest.getEmail());
            request.put("address", updateRequest.getAddress());    // 地址字段同步
            request.put("avatarUrl", updateRequest.getAvatar());    // 用户服务的avatar对应认证服务的avatarUrl
            
            HttpEntity<Map<String, Object>> entity = new HttpEntity<>(request, headers);
            
            // 调用认证服务的用户信息更新接口
            restTemplate.put(url, entity);
            
            log.info("认证服务数据库更新成功: {}", userId);
        } catch (Exception e) {
            log.error("调用认证服务更新用户信息失败: userId={}, error={}", userId, e.getMessage(), e);
            // 这里不抛出异常，因为用户服务已经更新成功，只记录日志
            // 在实际生产环境中，可能需要根据业务需求决定是否回滚用户服务的更新
        }
        
        return userMapper.entityToResponseDto(user);
    }

    @Override
    @Transactional
    public boolean changePassword(Long userId, ChangePasswordRequest changePasswordRequest) {
        log.info("修改用户密码: {}", userId);

        // 权限验证 - 确保用户只能修改自己的密码
        Authentication authentication = SecurityContextHolder.getContext().getAuthentication();
        if (authentication == null || !authentication.isAuthenticated() || authentication.getPrincipal().equals("anonymousUser")) {
            log.warn("未认证用户尝试修改密码，用户ID: {}", userId);
            throw new BusinessException("未认证，无法修改密码");
        }
        
        // 检查是否是管理员
        boolean isAdmin = authentication.getAuthorities().stream()
                .anyMatch(auth -> auth.getAuthority().equals("ROLE_ADMIN"));
        
        // 管理员可以修改任意用户密码
        if (isAdmin) {
            log.info("管理员修改密码，管理员: {}, 目标用户: {}", authentication.getName(), userId);
        } else {
            // 普通用户只能修改自己的密码
            // 从JWT令牌中获取用户ID
            String authenticatedUserId = null;
            
            // 尝试从JWT令牌中获取用户ID
            if (authentication.getDetails() instanceof Map) {
                @SuppressWarnings("unchecked")
                Map<String, Object> details = (Map<String, Object>) authentication.getDetails();
                authenticatedUserId = details.get("userId") != null ? details.get("userId").toString() : null;
            }
            
            // 如果从JWT令牌中获取不到，尝试从认证对象中获取
            if (authenticatedUserId == null) {
                authenticatedUserId = authentication.getPrincipal().toString();
            }
            
            log.info("从认证对象获取用户ID: {}, 请求修改的用户ID: {}", authenticatedUserId, userId);
            
            // 验证用户ID是否匹配
            try {
                long authenticatedUserIdLong = Long.parseLong(authenticatedUserId);
                if (authenticatedUserIdLong != userId) {
                    log.warn("权限验证失败: 认证用户ID {} 与请求修改的用户ID {} 不匹配", authenticatedUserId, userId);
                    throw new BusinessException("无权修改其他用户密码");
                }
            } catch (NumberFormatException e) {
                log.warn("权限验证失败: 认证用户ID {} 格式错误", authenticatedUserId);
                throw new BusinessException("无效的用户ID");
            }
            
            log.info("用户修改自己的密码，用户: {}, 用户ID: {}", authentication.getName(), userId);
        }

        // 验证新密码和确认密码是否一致
        if (!changePasswordRequest.getNewPassword().equals(changePasswordRequest.getConfirmPassword())) {
            throw new BusinessException("新密码和确认密码不一致");
        }

        // 直接调用认证服务修改密码
        try {
            String url = authServiceUrl + "/api/auth/change-password/" + userId;
            
            HttpHeaders headers = new HttpHeaders();
            headers.setContentType(MediaType.APPLICATION_JSON);
            
            Map<String, String> request = new HashMap<>();
            request.put("oldPassword", changePasswordRequest.getOldPassword());
            request.put("newPassword", changePasswordRequest.getNewPassword());
            request.put("confirmPassword", changePasswordRequest.getConfirmPassword());
            HttpEntity<Map<String, String>> entity = new HttpEntity<>(request, headers);
            
            // 调用认证服务的密码修改接口
            restTemplate.put(url, entity);
            
            log.info("用户密码修改成功: {}", userId);
            
            // 发布密码变更事件到Kafka
            try {
                User user = userRepository.findById(userId)
                        .orElseThrow(() -> new BusinessException("用户不存在"));
                userEventPublisher.publishPasswordChanged(user);
                log.info("密码变更事件已发布到Kafka - 用户ID: {}", userId);
            } catch (Exception eventException) {
                log.warn("密码变更事件发布失败 - 用户ID: {}, 错误: {}", userId, eventException.getMessage());
            }
            
            return true;
        } catch (Exception e) {
            log.error("调用认证服务修改密码失败: userId={}, error={}", userId, e.getMessage(), e);
            throw new BusinessException("密码修改失败: " + e.getMessage());
        }
    }

    @Override
    @Transactional
    public UserResponse updateAvatar(Long userId, String avatarUrl) {
        log.info("更新用户头像: {}", userId);

        // 权限验证 - 确保用户只能更新自己的头像
        Authentication authentication = SecurityContextHolder.getContext().getAuthentication();
        if (authentication != null && authentication.isAuthenticated() && !authentication.getPrincipal().equals("anonymousUser")) {
            String currentUsername = authentication.getName();
            
            // 检查是否是管理员
            boolean isAdmin = authentication.getAuthorities().stream()
                    .anyMatch(auth -> auth.getAuthority().equals("ROLE_ADMIN"));
            
            // 管理员可以更新任意用户头像
            if (isAdmin) {
                log.info("管理员更新头像，管理员: {}, 目标用户: {}", currentUsername, userId);
            } else {
                // 普通用户只能更新自己的头像
                // 从请求头中获取用户ID（网关传递的用户ID）
                HttpServletRequest request = ((ServletRequestAttributes) RequestContextHolder.currentRequestAttributes()).getRequest();
                String authenticatedUserId = request.getHeader("X-User-ID");
                
                if (authenticatedUserId != null) {
                    log.info("从请求头获取用户ID: {}, 请求更新的用户ID: {}", authenticatedUserId, userId);
                }
                
                // 验证用户ID是否匹配
                if (authenticatedUserId == null || !authenticatedUserId.equals(userId.toString())) {
                    log.warn("权限验证失败: 认证用户ID {} 与请求更新的用户ID {} 不匹配", authenticatedUserId, userId);
                    throw new BusinessException("无权更新其他用户头像");
                }
                
                log.info("用户更新自己的头像，用户: {}, 用户ID: {}", currentUsername, userId);
            }
        } else {
            log.warn("未认证用户尝试更新头像，用户ID: {}", userId);
            throw new BusinessException("未认证，无法更新头像");
        }

        User user = userRepository.findById(userId)
                .orElseThrow(() -> new BusinessException("用户不存在"));

        user.setAvatar(avatarUrl);
        user = userRepository.save(user);
        
        log.info("用户头像更新成功: {}", userId);
        return userMapper.entityToResponseDto(user);
    }

    @Override
    @Transactional
    public void syncUser(UserController.SyncUserRequest syncRequest) {
        try {
            log.info("开始同步用户数据: userId={}, username={}", syncRequest.getUserId(), syncRequest.getUsername());
            
            // 检查用户ID是否已存在
            if (userRepository.existsById(syncRequest.getUserId())) {
                log.info("用户ID已存在，更新用户信息: userId={}", syncRequest.getUserId());
                // 用户已存在，更新用户信息
                User user = userRepository.findById(syncRequest.getUserId())
                        .orElseThrow(() -> new BusinessException("用户不存在"));
                
                // 更新用户信息（处理字段名不匹配问题）
                user.setUsername(syncRequest.getUsername());
                user.setEmail(syncRequest.getEmail());
                // 处理phoneNumber与phone字段的映射
                user.setPhone(syncRequest.getPhoneNumber());
                // 处理地址字段
                user.setAddress(syncRequest.getAddress());
                // 处理nickName与nickname字段的映射
                user.setNickname(syncRequest.getNickName());
                user.setUpdatedAt(LocalDateTime.now());
                
                userRepository.save(user);
                log.info("用户信息更新成功: {}", user.getId());
            } else {
                log.info("用户ID不存在，创建新用户: userId={}", syncRequest.getUserId());
                // 检查用户名是否已存在
                if (userRepository.existsByUsername(syncRequest.getUsername())) {
                    log.warn("用户名已存在，跳过同步: {}", syncRequest.getUsername());
                    return;
                }

                // 检查邮箱是否已存在
                if (syncRequest.getEmail() != null && userRepository.existsByEmail(syncRequest.getEmail())) {
                    log.warn("邮箱已存在，跳过同步: {}", syncRequest.getEmail());
                    return;
                }
                
                // 用户不存在，创建新用户
                User user = new User();
                user.setId(syncRequest.getUserId()); // 使用认证服务的用户ID
                user.setUsername(syncRequest.getUsername());
                user.setEmail(syncRequest.getEmail());
                // 处理phoneNumber与phone字段的映射
                user.setPhone(syncRequest.getPhoneNumber());
                // 处理地址字段
                user.setAddress(syncRequest.getAddress());
                // 处理nickName与nickname字段的映射
                user.setNickname(syncRequest.getNickName());
                user.setPassword(""); // 密码在认证服务中管理，这里设置为空
                user.setAvatar(""); // 默认头像
                user.setCreatedAt(LocalDateTime.now());
                user.setUpdatedAt(LocalDateTime.now());
                // 设置默认角色 - 根据同步请求中的用户类型设置角色
                String userType = syncRequest.getUserType();
                Set<String> roles = new HashSet<>();
                if ("ADMIN".equals(userType)) {
                    roles.add("ROLE_ADMIN");
                } else {
                    roles.add("ROLE_USER");
                }
                user.setRoles(roles);
                
                userRepository.save(user);
                log.info("用户创建成功: {}", user.getId());
            }
        } catch (Exception e) {
            log.error("同步用户数据失败: userId={}, username={}, error={}", 
                     syncRequest.getUserId(), syncRequest.getUsername(), e.getMessage(), e);
            throw new BusinessException("用户数据同步失败: " + e.getMessage());
        }
    }

    /**
     * 生成服务间调用的认证令牌
     * @return 服务令牌
     */
    private String generateServiceToken() {
        try {
            // 使用服务账户调用认证服务获取令牌
            String url = authServiceUrl + "/api/auth/service-login";
            
            Map<String, String> request = new HashMap<>();
            request.put("serviceId", "user-service");
            request.put("serviceKey", "user-service-secret-key-12345");
            
            HttpHeaders headers = new HttpHeaders();
            headers.setContentType(MediaType.APPLICATION_JSON);
            HttpEntity<Map<String, String>> entity = new HttpEntity<>(request, headers);
            
            ResponseEntity<Map> response = restTemplate.postForEntity(url, entity, Map.class);
            
            if (response.getStatusCode().is2xxSuccessful() && response.getBody() != null) {
                Map<String, Object> responseBody = response.getBody();
                return (String) responseBody.get("token");
            }
            
            log.error("获取服务令牌失败，响应: {}", response);
            return null;
        } catch (Exception e) {
            log.error("生成服务令牌异常: {}", e.getMessage(), e);
            return null;
        }
    }

    @Override
    @Transactional
    public void syncUserFromAuthService(Long userId) {
        try {
            log.info("从认证服务同步用户数据，用户ID: {}", userId);
            
            // 调用认证服务获取用户信息
            String url = authServiceUrl + "/api/auth/user/" + userId;
            
            // 设置请求头，添加认证信息
            HttpHeaders headers = new HttpHeaders();
            headers.setContentType(MediaType.APPLICATION_JSON);
            // 添加服务间调用的认证令牌
            headers.set("Authorization", "Bearer " + generateServiceToken());
            
            HttpEntity<String> entity = new HttpEntity<>(headers);
            
            ResponseEntity<Map> response = restTemplate.exchange(url, HttpMethod.GET, entity, Map.class);
            
            if (!response.getStatusCode().is2xxSuccessful() || response.getBody() == null) {
                log.error("从认证服务获取用户信息失败: userId={}, response={}", userId, response);
                throw new BusinessException("从认证服务获取用户信息失败");
            }
            
            Map<String, Object> authUserData = response.getBody();
            
            // 创建同步请求对象
            UserController.SyncUserRequest syncRequest = new UserController.SyncUserRequest();
            syncRequest.setUserId(userId);
            syncRequest.setUsername((String) authUserData.get("username"));
            syncRequest.setEmail((String) authUserData.get("email"));
            syncRequest.setPhoneNumber((String) authUserData.get("phoneNumber"));
            syncRequest.setAddress((String) authUserData.get("address"));
            syncRequest.setNickName((String) authUserData.get("nickName"));
            syncRequest.setUserType((String) authUserData.get("userType"));
            
            log.info("从认证服务获取的用户数据: userId={}, username={}, email={}, phoneNumber={}, address={}, nickName={}", 
                    userId, syncRequest.getUsername(), syncRequest.getEmail(), 
                    syncRequest.getPhoneNumber(), syncRequest.getAddress(), syncRequest.getNickName());
            
            // 调用现有的同步方法
            syncUser(syncRequest);
            
            log.info("从认证服务同步用户数据成功，用户ID: {}", userId);
        } catch (Exception e) {
            log.error("从认证服务同步用户数据失败: userId={}, error={}", userId, e.getMessage(), e);
            throw new BusinessException("从认证服务同步用户数据失败: " + e.getMessage());
        }
    }

    @Override
    @Transactional
    public void syncPassword(Long userId, String encodedPassword) {
        try {
            log.info("同步用户密码: userId={}", userId);
            
            User user = userRepository.findById(userId)
                    .orElseThrow(() -> new BusinessException("用户不存在"));
            
            // 更新密码（认证服务已加密的密码）
            user.setPassword(encodedPassword);
            user.setUpdatedAt(LocalDateTime.now());
            
            userRepository.save(user);
            log.info("用户密码同步成功: {}", userId);
        } catch (Exception e) {
            log.error("同步用户密码失败: userId={}, error={}", userId, e.getMessage(), e);
            throw new BusinessException("用户密码同步失败: " + e.getMessage());
        }
    }

    /**
     * 同步密码到认证服务
     * @param userId 用户ID
     * @param encodedPassword 已加密的密码
     */
    private void syncPasswordToAuthService(Long userId, String encodedPassword) {
        try {
            String url = authServiceUrl + "/api/auth/sync-password/" + userId;
            
            HttpHeaders headers = new HttpHeaders();
            headers.setContentType(MediaType.APPLICATION_JSON);
            
            Map<String, String> request = new HashMap<>();
            request.put("encodedPassword", encodedPassword);
            HttpEntity<Map<String, String>> entity = new HttpEntity<>(request, headers);
            
            // 调用认证服务的密码同步接口
            restTemplate.postForObject(url, entity, String.class);
            
            log.info("密码同步到认证服务成功: userId={}", userId);
        } catch (Exception e) {
            log.error("密码同步到认证服务失败: userId={}, error={}", userId, e.getMessage(), e);
            throw new BusinessException("密码同步到认证服务失败: " + e.getMessage());
        }
    }

    // ===== 管理员相关方法实现 =====

    @Override
    public Page<UserResponse> getUsers(String keyword, String userType, Boolean status, Pageable pageable) {
        log.info("管理员获取用户列表，关键词: {}, 用户类型: {}, 状态: {}", keyword, userType, status);
        
        // 构建查询条件
        Page<User> userPage;
        if (keyword != null && !keyword.trim().isEmpty()) {
            // 根据关键词搜索用户名、邮箱或手机号
            userPage = userRepository.findByUsernameContainingOrEmailContainingOrPhoneContaining(
                    keyword, keyword, keyword, pageable);
        } else if (userType != null && !userType.trim().isEmpty()) {
            // 根据用户类型筛选
            userPage = userRepository.findByUserType(userType, pageable);
        } else if (status != null) {
            // 根据状态筛选
            userPage = userRepository.findByEnabled(status, pageable);
        } else {
            // 获取所有用户
            userPage = userRepository.findAll(pageable);
        }
        
        // 转换为响应DTO
        List<UserResponse> userResponses = userPage.getContent().stream()
                .map(userMapper::entityToResponseDto)
                .collect(Collectors.toList());
        
        return new org.springframework.data.domain.PageImpl<>(userResponses, pageable, userPage.getTotalElements());
    }

    @Override
    @Transactional
    public UserResponse createUser(UpdateUserRequest userRequest) {
        log.info("管理员创建用户: {}", userRequest.getUsername());
        
        // 检查用户名是否已存在
        if (userRepository.existsByUsername(userRequest.getUsername())) {
            throw new BusinessException(400, "用户名已存在");
        }
        
        // 检查邮箱是否已存在
        if (userRepository.existsByEmail(userRequest.getEmail())) {
            throw new BusinessException(400, "邮箱已被注册");
        }
        
        // 首先调用认证服务注册接口
        try {
            Map<String, Object> authRegisterRequest = new HashMap<>();
            authRegisterRequest.put("username", userRequest.getUsername());
            authRegisterRequest.put("password", userRequest.getPassword() != null && !userRequest.getPassword().isEmpty() 
                ? userRequest.getPassword() 
                : "default123"); // 默认密码
            authRegisterRequest.put("email", userRequest.getEmail());

            // 调用认证服务注册接口
            Map<String, Object> authResponse;
            try {
                authResponse = webClient.post()
                        .uri("/api/auth/register")
                        .contentType(MediaType.APPLICATION_JSON)
                        .bodyValue(authRegisterRequest)
                        .retrieve()
                        .bodyToMono(new ParameterizedTypeReference<Map<String, Object>>() {})
                        .block();
            } catch (Exception e) {
                log.error("调用认证服务注册接口失败: {}", e.getMessage());
                throw new BusinessException("用户创建失败，认证服务不可用");
            }

            log.info("认证服务注册成功: {}", authResponse);
            
            // 从认证服务响应中获取用户ID
            Long authUserId = null;
            if (authResponse != null && authResponse.containsKey("userId")) {
                authUserId = Long.valueOf(authResponse.get("userId").toString());
            }
            
            if (authUserId == null) {
                log.error("认证服务未返回用户ID: {}", userRequest.getUsername());
                throw new BusinessException("用户创建失败，无法获取用户ID");
            }
            
            // 认证服务注册成功后，在用户服务中创建用户记录
            User user = new User();
            user.setId(authUserId); // 使用认证服务返回的用户ID
            user.setUsername(userRequest.getUsername());
            user.setPassword(""); // 密码由认证服务管理，用户服务不需要存储
            user.setEmail(userRequest.getEmail());
            user.setPhone(userRequest.getPhone());
            user.setNickname(userRequest.getNickname());
            user.setAvatar(userRequest.getAvatar());
            user.setUserType(userRequest.getUserType() != null ? userRequest.getUserType() : "USER");
            user.setEnabled(userRequest.getEnabled() != null ? userRequest.getEnabled() : true);
            user.setCreatedAt(LocalDateTime.now());
            user.setUpdatedAt(LocalDateTime.now());
            
            // 设置角色 - 使用可变的HashSet避免UnsupportedOperationException
            Set<String> roles = new HashSet<>();
            if ("ADMIN".equals(userRequest.getUserType())) {
                roles.add("ROLE_ADMIN");
            } else {
                roles.add("ROLE_USER");
            }
            user.setRoles(roles);

            // 保存用户并确保创建成功
            user = userRepository.save(user);
            
            // 验证用户是否成功创建
            if (user.getId() == null) {
                log.error("用户服务数据创建失败，用户ID为空: {}", userRequest.getUsername());
                throw new BusinessException("用户数据创建失败，请重试");
            }
            
            // 双重确认用户数据已持久化
            final String username = userRequest.getUsername();
            final Long userId = user.getId();
            User savedUser = userRepository.findById(userId)
                    .orElseThrow(() -> {
                        log.error("用户数据保存后验证失败: userId={}, username={}", 
                                 userId, username);
                        return new BusinessException("用户数据保存验证失败");
                    });
            
            log.info("用户服务创建成功: {}, userId: {}", savedUser.getUsername(), savedUser.getId());

            // 发送用户创建成功事件到Kafka
            try {
                // 使用用户事件发布器发送事件
                userEventPublisher.publishUserCreated(savedUser);
                log.info("用户创建事件已发送到Kafka - 用户ID: {}, 用户名: {}", savedUser.getId(), savedUser.getUsername());
                
            } catch (Exception e) {
                log.error("用户创建事件发送失败 - 用户ID: {}, 错误: {}", savedUser.getId(), e.getMessage(), e);
                
                // 清理已创建的用户数据
                try {
                    userRepository.delete(savedUser);
                    log.info("创建失败，已清理用户数据 - 用户ID: {}", savedUser.getId());
                } catch (Exception cleanupEx) {
                    log.error("清理用户数据失败 - 用户ID: {}, 错误: {}", savedUser.getId(), cleanupEx.getMessage());
                }
                
                // 回滚认证服务注册
                try {
                    webClient.delete()
                            .uri("/api/auth/user/" + username)
                            .retrieve()
                            .bodyToMono(Void.class)
                            .onErrorResume(rollbackError -> {
                                log.error("回滚认证服务注册失败: username={}, error={}", 
                                         username, rollbackError.getMessage());
                                return Mono.empty();
                            })
                            .block();
                    
                    log.info("认证服务注册回滚成功: {}", username);
                } catch (Exception rollbackEx) {
                    log.error("认证服务注册回滚失败: username={}, error={}", 
                             username, rollbackEx.getMessage(), rollbackEx);
                }
                
                throw new BusinessException("用户创建失败: 事件发送失败");
            }

            return userMapper.entityToResponseDto(savedUser);
            
        } catch (Exception e) {
            log.error("用户创建过程失败: {}", e.getMessage(), e);
            
            // 如果是用户服务数据创建失败，需要回滚认证服务注册
            if (e.getMessage() != null && (e.getMessage().contains("用户数据创建失败") || 
                                         e.getMessage().contains("用户数据保存验证失败"))) {
                log.warn("用户服务数据创建失败，准备回滚认证服务注册: {}", userRequest.getUsername());
                
                try {
                    // 调用认证服务删除用户接口进行回滚
                    final String rollbackUsername = userRequest.getUsername();
                    webClient.delete()
                            .uri("/api/auth/user/" + rollbackUsername)
                            .retrieve()
                            .bodyToMono(Void.class)
                            .onErrorResume(rollbackError -> {
                                log.error("回滚认证服务注册失败: username={}, error={}", 
                                         rollbackUsername, rollbackError.getMessage());
                                return Mono.empty();
                            })
                            .block();
                    
                    log.info("认证服务注册回滚成功: {}", rollbackUsername);
                } catch (Exception rollbackEx) {
                    log.error("认证服务注册回滚失败: username={}, error={}", 
                             userRequest.getUsername(), rollbackEx.getMessage(), rollbackEx);
                }
                
                throw new BusinessException("用户创建失败，请重试");
            }
            
            // 其他异常处理
            if (e.getMessage() != null && e.getMessage().contains("用户名已存在")) {
                throw new BusinessException(400, "用户名已存在");
            } else if (e.getMessage() != null && e.getMessage().contains("邮箱已被注册")) {
                throw new BusinessException(400, "邮箱已被注册");
            } else {
                throw new BusinessException("用户创建失败: " + e.getMessage());
            }
        }
    }

    @Override
    @Transactional
    public UserResponse updateUser(Long userId, UpdateUserRequest userRequest) {
        log.info("管理员更新用户: {}", userId);
        
        User user = userRepository.findById(userId)
                .orElseThrow(() -> new BusinessException("用户不存在"));
        
        // 检查用户名是否被其他用户使用
        if (!user.getUsername().equals(userRequest.getUsername()) && 
            userRepository.existsByUsername(userRequest.getUsername())) {
            throw new BusinessException("用户名已存在");
        }
        
        // 检查邮箱是否被其他用户使用
        if (!user.getEmail().equals(userRequest.getEmail()) && 
            userRepository.existsByEmail(userRequest.getEmail())) {
            throw new BusinessException("邮箱已被注册");
        }
        
        // 更新用户信息
        user.setUsername(userRequest.getUsername());
        user.setEmail(userRequest.getEmail());
        user.setPhone(userRequest.getPhone());
        user.setNickname(userRequest.getNickname());
        user.setAvatar(userRequest.getAvatar());
        user.setUserType(userRequest.getUserType());
        user.setEnabled(userRequest.getEnabled());
        user.setUpdatedAt(LocalDateTime.now());
        
        // 更新角色 - 使用可变的HashSet避免UnsupportedOperationException
        Set<String> roles = new HashSet<>();
        if ("ADMIN".equals(userRequest.getUserType())) {
            roles.add("ROLE_ADMIN");
        } else if ("SUPER_ADMIN".equals(userRequest.getUserType())) {
            roles.add("ROLE_ADMIN");  // 超级管理员也拥有管理员角色
        } else {
            roles.add("ROLE_USER");
        }
        user.setRoles(roles);
        
        user = userRepository.save(user);
        
        // 发布用户更新事件到Kafka - 保证数据最终一致性
        try {
            userEventPublisher.publishUserUpdated(user);
            log.info("用户更新事件已发布到Kafka - 用户ID: {}, 用户类型: {}", userId, user.getUserType());
        } catch (Exception eventException) {
            log.warn("用户更新事件发布失败 - 用户ID: {}, 错误: {}", userId, eventException.getMessage());
            // 事件发布失败不回滚事务，记录日志即可
        }
        
        // 异步同步到认证服务 - 使用更新用户资料接口，避免阻塞前端请求
        try {
            Map<String, Object> authRequest = new HashMap<>();
            authRequest.put("username", user.getUsername());
            authRequest.put("email", user.getEmail());
            
            // 保存用户名到final变量，用于lambda表达式
            final String username = user.getUsername();
            
            // 使用异步方式，不等待响应立即返回
            webClient.put()
                    .uri("/api/auth/update-profile/" + user.getId())
                    .contentType(MediaType.APPLICATION_JSON)
                    .bodyValue(authRequest)
                    .retrieve()
                    .bodyToMono(Map.class)
                    .subscribe(
                        response -> log.info("用户信息同步到认证服务成功: {}", username),
                        error -> log.error("用户信息同步到认证服务失败: {}", username, error)
                    );
                    
            log.info("开始异步同步用户信息到认证服务: {}", user.getUsername());
        } catch (Exception e) {
            log.error("用户信息同步到认证服务失败: {}", user.getUsername(), e);
            // 不回滚，因为用户服务已经更新成功
        }
        
        return userMapper.entityToResponseDto(user);
    }

    @Override
    @Transactional
    public boolean deleteUser(Long userId) {
        log.info("管理员删除用户: {}", userId);
        
        // 验证当前用户是否有管理员权限
        Authentication authentication = SecurityContextHolder.getContext().getAuthentication();
        if (authentication == null || !authentication.getAuthorities().contains(new SimpleGrantedAuthority("ROLE_ADMIN"))) {
            log.error("删除用户失败: 当前用户没有管理员权限");
            throw new BusinessException("没有权限删除用户");
        }
        
        // 检查用户ID是否有效
        if (userId == null || userId <= 0) {
            log.error("删除用户失败: 无效的用户ID: {}", userId);
            throw new BusinessException("无效的用户ID");
        }
        
        User user = userRepository.findById(userId)
                .orElseThrow(() -> new BusinessException("用户不存在，ID: " + userId));
        
        // 防止管理员删除自己
        Long currentUserId = null;
        try {
            currentUserId = Long.parseLong(authentication.getName());
        } catch (NumberFormatException e) {
            // 如果无法解析，尝试从认证主体获取
            if (authentication.getPrincipal() instanceof Long) {
                currentUserId = (Long) authentication.getPrincipal();
            }
        }
        
        if (currentUserId != null && currentUserId.equals(userId)) {
            log.error("删除用户失败: 不能删除自己的账户");
            throw new BusinessException("不能删除自己的账户");
        }
        
        // 保存用户信息用于事件发布
        String username = user.getUsername();
        String email = user.getEmail();
        
        try {
            userRepository.delete(user);
            log.info("用户从用户服务删除成功: {}", username);
        } catch (Exception e) {
            log.error("从用户服务删除用户失败: {}", username, e);
            throw new BusinessException("删除用户失败: " + e.getMessage());
        }
        
        // 同步删除到认证服务
        try {
            webClient.delete()
                    .uri("/api/auth/users/" + userId)
                    .retrieve()
                    .bodyToMono(Map.class)
                    .block();
                    
            log.info("用户从认证服务删除成功: {}", username);
        } catch (Exception e) {
            log.error("用户从认证服务删除失败: {}", username, e);
            // 不回滚，因为用户服务已经删除成功
        }
        
        // 发送用户删除事件到Kafka
        try {
            UserDeletedEvent event = new UserDeletedEvent(
                    userId,
                    username,
                    "USER_SERVICE",
                    "用户服务删除用户",
                    "USER_SERVICE"
            );
            
            userDeletedEventProducer.sendUserDeletedEvent(event);
            log.info("用户删除事件发送到Kafka成功 - 用户ID: {}, 用户名: {}", userId, username);
        } catch (Exception e) {
            log.error("用户删除事件发送到Kafka失败 - 用户ID: {}, 用户名: {}, 错误: {}", userId, username, e.getMessage(), e);
            // 不回滚，因为用户服务已经删除成功
        }
        
        return true;
    }

    @Override
    @Transactional
    public boolean batchDeleteUsers(List<Long> userIds) {
        log.info("管理员批量删除用户: {}", userIds);
        
        List<User> users = userRepository.findAllById(userIds);
        userRepository.deleteAll(users);
        
        // 同步批量删除到认证服务
        try {
            Map<String, Object> authRequest = new HashMap<>();
            authRequest.put("ids", userIds);
            
            webClient.post()
                    .uri("/api/auth/users/batch-delete")
                    .contentType(MediaType.APPLICATION_JSON)
                    .bodyValue(authRequest)
                    .retrieve()
                    .bodyToMono(Map.class)
                    .block();
                    
            log.info("用户批量删除同步到认证服务成功: {} 个用户", users.size());
        } catch (Exception e) {
            log.error("用户批量删除同步到认证服务失败: {}", e.getMessage(), e);
            // 不回滚，因为用户服务已经删除成功
        }
        
        // 发送批量用户删除事件到Kafka
        try {
            for (User user : users) {
                try {
                    UserDeletedEvent event = new UserDeletedEvent(
                            user.getId(),
                            user.getUsername(),
                            "USER_SERVICE",
                            "用户服务批量删除用户",
                            "USER_SERVICE"
                    );
                    
                    userDeletedEventProducer.sendUserDeletedEvent(event);
                    log.info("用户删除事件发送到Kafka成功 - 用户ID: {}, 用户名: {}", user.getId(), user.getUsername());
                } catch (Exception e) {
                    log.error("单个用户删除事件发送到Kafka失败 - 用户ID: {}, 用户名: {}, 错误: {}", 
                             user.getId(), user.getUsername(), e.getMessage(), e);
                    // 继续处理其他用户，不回滚
                }
            }
            log.info("批量用户删除事件发送到Kafka完成 - 共 {} 个用户", users.size());
        } catch (Exception e) {
            log.error("批量用户删除事件发送到Kafka失败 - 错误: {}", e.getMessage(), e);
            // 不回滚，因为用户服务已经删除成功
        }
        
        return true;
    }

    @Override
    @Transactional
    public boolean updateUserStatus(Long userId, boolean enabled) {
        log.info("管理员更新用户状态: {}, 状态: {}", userId, enabled);
        
        User user = userRepository.findById(userId)
                .orElseThrow(() -> new BusinessException("用户不存在"));
        
        user.setEnabled(enabled);
        user.setUpdatedAt(LocalDateTime.now());
        userRepository.save(user);
        
        // 同步状态到认证服务
        try {
            Map<String, Object> authRequest = new HashMap<>();
            authRequest.put("userId", userId);
            authRequest.put("enabled", enabled);
            
            webClient.put()
                    .uri("/api/auth/users/" + userId + "/status")
                    .contentType(MediaType.APPLICATION_JSON)
                    .bodyValue(authRequest)
                    .retrieve()
                    .bodyToMono(Map.class)
                    .block();
                    
            log.info("用户状态同步到认证服务成功: {}, 状态: {}", user.getUsername(), enabled);
        } catch (Exception e) {
            log.error("用户状态同步到认证服务失败: {}", user.getUsername(), e);
            // 不回滚，因为用户服务已经更新成功
        }
        
        return true;
    }

    @Override
    @Transactional
    public boolean resetUserPassword(Long userId) {
        log.info("管理员重置用户密码: {}", userId);
        
        User user = userRepository.findById(userId)
                .orElseThrow(() -> new BusinessException("用户不存在"));
        
        // 生成默认密码
        String defaultPassword = "123456";
        String encodedPassword = passwordEncoder.encode(defaultPassword);
        
        // 同步到认证服务
        try {
            Map<String, Object> authRequest = new HashMap<>();
            authRequest.put("userId", userId);
            authRequest.put("encodedPassword", encodedPassword);
            
            webClient.put()
                    .uri("/api/auth/users/" + userId + "/reset-password")
                    .contentType(MediaType.APPLICATION_JSON)
                    .bodyValue(authRequest)
                    .retrieve()
                    .bodyToMono(Map.class)
                    .block();
                    
            log.info("用户密码重置同步到认证服务成功: {}", user.getUsername());
        } catch (Exception e) {
            log.error("用户密码重置同步到认证服务失败: {}", user.getUsername(), e);
            throw new BusinessException("密码重置失败: " + e.getMessage());
        }
        
        return true;
    }

    @Override
    public UserStatsResponse getUserStats() {
        log.info("获取用户统计信息");
        
        long totalUsers = userRepository.count();
        long enabledUsers = userRepository.countByEnabled(true);
        long disabledUsers = userRepository.countByEnabled(false);
        long normalUsers = userRepository.countByUserType("USER");
        long adminUsers = userRepository.countByUserType("ADMIN") + userRepository.countByUserType("SUPER_ADMIN");
        
        // 获取今日新增用户（假设有createdAt字段）
        LocalDateTime todayStart = LocalDateTime.now().withHour(0).withMinute(0).withSecond(0);
        long todayNewUsers = userRepository.countByCreatedAtAfter(todayStart);
        
        // 获取本周新增用户
        LocalDateTime weekStart = LocalDateTime.now().with(java.time.DayOfWeek.MONDAY).withHour(0).withMinute(0).withSecond(0);
        long weekNewUsers = userRepository.countByCreatedAtAfter(weekStart);
        
        // 获取本月新增用户
        LocalDateTime monthStart = LocalDateTime.now().withDayOfMonth(1).withHour(0).withMinute(0).withSecond(0);
        long monthNewUsers = userRepository.countByCreatedAtAfter(monthStart);
        
        return UserStatsResponse.builder()
                .totalUsers(totalUsers)
                .todayNewUsers(todayNewUsers)
                .weekNewUsers(weekNewUsers)
                .monthNewUsers(monthNewUsers)
                .activeUsers(enabledUsers) // 假设启用的用户就是活跃用户
                .normalUsers(normalUsers)
                .vipUsers(0L) // 暂时不支持VIP用户
                .adminUsers(adminUsers)
                .enabledUsers(enabledUsers)
                .disabledUsers(disabledUsers)
                .build();
    }

    /**
     * 检查用户名是否存在
     * 
     * @param username 用户名
     * @return boolean 用户存在性检查结果
     */
    @Override
    public boolean existsByUsername(String username) {
        log.info("检查用户名是否存在: {}", username);
        
        try {
            boolean exists = userRepository.existsByUsername(username);
            log.info("用户名存在性检查结果 - 用户名: {}, 存在: {}", username, exists);
            return exists;
        } catch (Exception e) {
            log.error("检查用户名存在性失败 - 用户名: {}, 错误: {}", username, e.getMessage(), e);
            return false;
        }
    }

    /**
     * 根据用户名删除用户（用于事件同步）
     * 
     * @param username 用户名
     * @return boolean 是否删除成功
     */
    @Override
    @Transactional
    public boolean deleteUserByUsername(String username) {
        log.info("根据用户名删除用户 - 用户名: {}", username);
        
        try {
            // 查找用户
            User user = userRepository.findByUsername(username)
                .orElse(null);
            
            if (user == null) {
                log.warn("用户不存在 - 用户名: {}", username);
                return false;
            }
            
            // 保存用户删除前的信息（用于事件发布）
            UserDeletedEvent userDeletedEvent = new UserDeletedEvent(
                user.getId(),
                user.getUsername(),
                "USER_SERVICE",
                "用户服务事件同步删除",
                "USER_SERVICE"
            );
            
            // 删除用户
            userRepository.delete(user);
            
            // 发布用户删除事件
            try {
                userDeletedEventProducer.sendUserDeletedEvent(userDeletedEvent);
                log.info("用户删除事件发布成功 - 用户ID: {}, 用户名: {}", user.getId(), user.getUsername());
            } catch (Exception e) {
                log.error("用户删除事件发布失败 - 用户ID: {}, 错误: {}", user.getId(), e.getMessage(), e);
                // 事件发布失败不影响用户删除操作
            }
            
            log.info("用户删除成功 - 用户ID: {}, 用户名: {}", user.getId(), user.getUsername());
            return true;
            
        } catch (Exception e) {
            log.error("用户删除失败 - 用户名: {}, 错误: {}", username, e.getMessage(), e);
            return false;
        }
    }
}